package com.wxzhou.wangzhuan.service;

import com.google.gson.Gson;
import com.wxzhou.wangzhuan.entity.AdLog;
import io.quarkus.redis.datasource.ReactiveRedisDataSource;
import io.quarkus.redis.datasource.keys.ReactiveKeyCommands;
import io.quarkus.redis.datasource.list.ReactiveListCommands;
import io.quarkus.redis.datasource.stream.StreamCommands;
import io.smallrye.mutiny.Uni;
import jakarta.enterprise.context.ApplicationScoped;
import jakarta.inject.Inject;
import org.jboss.logging.Logger;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;

@ApplicationScoped
public class AdLogRedisService {

    private static final Logger LOG = Logger.getLogger(AdLogRedisService.class);

    @Inject
    ReactiveRedisDataSource reactiveRedisDataSource;

    private static final Gson GSON = new Gson();
    private static final String REDIS_KEY_PREFIX = "ad:log:";
    private static final Duration DEFAULT_EXPIRE = Duration.ofMinutes(5);

    private ReactiveListCommands<String, String> listCommands() {
        return reactiveRedisDataSource.list(String.class);
    }

    private ReactiveKeyCommands<String> keyCommands() {
        return reactiveRedisDataSource.key(String.class);
    }

    /**
     * 初始化或追加AdLog列表到Redis
     */
    public Uni<Void> appendAdLogs(String adidsn, List<AdLog> adLogs) {
        String redisKey = REDIS_KEY_PREFIX + adidsn;


        return Uni.combine().all().unis(
                // 批量添加AdLog到List
                listCommands().rpush(redisKey,
                        adLogs.stream()
                                .map(GSON::toJson)
                                .toArray(String[]::new)
                ),

                // 设置或更新过期时间（基于bs_ac值）
                handleExpirationByBsAc(redisKey, adLogs)
        ).discardItems();
    }

    /**
     * 追加单个AdLog到Redis
     */
    public Uni<Void> appendAdLog(String adidsn, AdLog adLog) {
        String redisKey = REDIS_KEY_PREFIX + adidsn;

        return Uni.combine().all().unis(
                // 添加单个AdLog到List
                listCommands().rpush(redisKey, GSON.toJson(adLog)),

                // 处理过期时间
                handleExpirationByBsAc(redisKey, List.of(adLog))
        ).discardItems();
    }

    /**
     * 根据bs_ac值处理过期时间
     */
    private Uni<Void> handleExpirationByBsAc(String redisKey, List<AdLog> adLogs) {
        return Uni.createFrom().item(adLogs)
                .onItem().transformToUni(logs -> {
                    for (AdLog log : logs) {
                        Integer bsAc = log.getBs_ac();
                        if (bsAc == null) {
                            continue;
                        }

//                        if (bsAc == 2 || bsAc == 3) {
//                            // bs_ac=2或3时删除整个缓存
//                            return keyCommands().del(redisKey)
//                                    .onItem().ignore().andContinueWithNull();
//                        } else if (bsAc == 0) {
//                            // bs_ac=0时设置10分钟过期
//                            return keyCommands().expire(redisKey, DEFAULT_EXPIRE)
//                                    .onItem().ignore().andContinueWithNull();
//                        }
                        return keyCommands().expire(redisKey, DEFAULT_EXPIRE)
                                .onItem().ignore().andContinueWithNull();

                    }
                    return Uni.createFrom().nullItem();
                });
    }

    /**
     * 获取指定adidsn的所有AdLog
     */
    public Uni<List<AdLog>> getAllAdLogs(String adidsn) {
        String redisKey = REDIS_KEY_PREFIX + adidsn;
        LOG.infof("1. getAllAdLogs started for adidsn: %s, redisKey: %s", adidsn, redisKey);

        return listCommands().lrange(redisKey, 0, -1)
                .onSubscription().invoke(() ->
                        LOG.infof("2. Redis lrange subscription started for key: %s", redisKey)
                )
                .onItem().transform(list -> {
                    List<AdLog> result = new ArrayList<>();
                    for (String adLog : list) {
                        result.add(GSON.fromJson(adLog, AdLog.class));
                    }
                    return result;
                })
                .onFailure().invoke(throwable ->
                        LOG.errorf("6. Error in getAllAdLogs for adidsn %s: %s", adidsn, throwable.getMessage())
                )
                .onFailure().recoverWithUni(throwable -> {
                    LOG.errorf("7. Recovering from error in getAllAdLogs", throwable);
                    return Uni.createFrom().failure(throwable);
                });
    }

    /**
     * 检查缓存是否存在
     */
    public Uni<Boolean> exists(String adidsn) {
        String redisKey = REDIS_KEY_PREFIX + adidsn;
        return keyCommands().exists(redisKey);
    }

    /**
     * 删除指定adidsn的缓存
     */
    public Uni<Integer> deleteAdLogs(String adidsn) {
        String redisKey = REDIS_KEY_PREFIX + adidsn;
        return keyCommands().del(redisKey);
    }

    /**
     * 获取缓存数量
     */
    public Uni<Long> getAdLogsCount(String adidsn) {
        String redisKey = REDIS_KEY_PREFIX + adidsn;
        return listCommands().llen(redisKey);
    }

    /**
     * 手动设置过期时间
     */
    public Uni<Boolean> setExpire(String adidsn, Duration duration) {
        String redisKey = REDIS_KEY_PREFIX + adidsn;
        return keyCommands().expire(redisKey, duration);
    }
}